package com.central.scheduler.api.service.impl;

import com.central.common.model.CodeEnum;
import com.central.common.model.Result;
import com.central.common.model.SysUser;
import com.central.common.utils.UserRoleUtil;
import com.central.scheduler.api.enums.ExecuteType;
import com.central.scheduler.api.enums.Status;
import com.central.scheduler.api.service.IExecutorService;
import com.central.scheduler.api.service.IMonitorService;
import com.central.scheduler.common.Constants;
import com.central.scheduler.dao.model.ProcessInstance;
import com.central.scheduler.remote.command.StateEventChangeCommand;
import com.central.scheduler.remote.processor.StateEventCallbackService;
import com.central.scheduler.service.process.ProcessService;
import com.central.scheduler.service.quartz.cron.CronUtils;
import com.central.scheduler.common.enums.*;
import com.central.scheduler.common.utils.DateUtils;
import com.central.scheduler.common.utils.JSONUtils;
import com.central.scheduler.dao.entity.Server;
import com.central.scheduler.dao.mapper.ProcessDefinitionMapper;
import com.central.scheduler.dao.model.Command;
import com.central.scheduler.dao.model.ProcessDefinition;
import com.central.scheduler.dao.model.Schedule;
import com.fasterxml.jackson.core.type.TypeReference;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.text.MessageFormat;
import java.util.*;

import static com.central.scheduler.common.Constants.*;

/**
 * @author Tindy
 * @date 2021/12/15
 * @describe
 */
@Service
public class ExecutorServiceImpl implements IExecutorService {

    private static final Logger logger = LoggerFactory.getLogger(ExecutorServiceImpl.class);

    @Autowired
    private ProcessDefinitionMapper processDefinitionMapper;

    @Autowired
    private IMonitorService monitorService;

    @Autowired
    private ProcessService processService;
    @Autowired
    StateEventCallbackService stateEventCallbackService;

    /**
     * execute process instance
     *
     * @param user
     * @param processDefinitionCode process definition code
     * @param cronTime cron time
     * @param commandType command type
     * @param failureStrategy failure strategy
     * @param startNodeList start nodelist
     * @param taskDependType node dependency type
     * @param warningType warning type
     * @param warningGroupId notify group id
     * @param processInstancePriority process instance priority
     * @param runMode run mode
     * @param timeout timeout
     * @param startParams the global param values which pass to new process instance
     * @param expectedParallelismNumber the expected parallelism number when execute complement in parallel mode
     * @return execute process instance code
     */
    @Override
    public Result execProcessInstance(SysUser user, Long processDefinitionCode, String cronTime, CommandType commandType,
                                      FailureStrategy failureStrategy, String startNodeList,
                                      TaskDependType taskDependType, WarningType warningType, Long warningGroupId,
                                      RunMode runMode, Priority processInstancePriority, Integer timeout,
                                      Map<String, String> startParams, Integer expectedParallelismNumber, Integer dryRun) {
        // check process define release state
        ProcessDefinition processDefinition = processDefinitionMapper.queryByCode(processDefinitionCode,user.getCurrentProjectId());
        Result result = checkProcessDefinitionValid(processDefinition);
        if (result.getResp_code() != CodeEnum.SUCCESS.getCode()) {
            return result;
        }
        //check user current project is process project
        if(!processDefinition.getProId().equals(user.getCurrentProjectId())) return Result.failed(Status.USER_NO_OPERATION_PERM.getMsg());

        // timeout is invalid
        if (timeout <= 0 || timeout > MAX_TASK_TIMEOUT) {
            return  Result.failed(Status.TASK_TIMEOUT_PARAMS_ERROR.getMsg());
        }

        // check master exists
        if (!checkMasterExists()) {
            return Result.failed(Status.MASTER_NOT_EXISTS.getMsg());
        }

        /**
         * create command
         */
        int create = this.createCommand(user, commandType, processDefinition.getCode(),
                taskDependType, failureStrategy, startNodeList, cronTime, warningType,
                warningGroupId, runMode, processInstancePriority, startParams, expectedParallelismNumber, dryRun);

        if (create > 0) {
            processDefinition.setWarningGroupId(warningGroupId);
            processDefinitionMapper.updateById(processDefinition);
            return Result.succeed(Status.SUCCESS);
        } else {
            return Result.failed(Status.START_PROCESS_INSTANCE_ERROR.getMsg());
        }
    }
    /**
     * check whether the process definition can be executed
     *
     *
     * @param processDefinition process definition
     * @return check result
     */
    @Override
    public Result checkProcessDefinitionValid(ProcessDefinition processDefinition) {
        if (processDefinition == null) {
            // check process definition exists
            return Result.failed(MessageFormat.format(Status.PROCESS_DEFINE_NOT_EXIST.getMsg(),processDefinition.getName()));
        } else if (processDefinition.getReleaseState() != ReleaseState.ONLINE) {
            // check process definition online
            return Result.failed(MessageFormat.format(Status.PROCESS_DEFINE_NOT_RELEASE.getMsg(),processDefinition.getName()));
        } else{
            return Result.succeed(Status.SUCCESS.getMsg());
        }
    }

    /**
     * do action to process instance：pause, stop, repeat, recover from pause, recover from stop
     * @param user
     * @param processInstanceId
     * @param executeType
     * @return
     */
    @Override
    public Result execute(SysUser user, Long processInstanceId, ExecuteType executeType) {

        // check master exists
        if (!checkMasterExists()) {
            return Result.failed(Status.MASTER_NOT_EXISTS.getMsg());
        }

        ProcessInstance processInstance = processService.findProcessInstanceDetailById(processInstanceId);
        if (processInstance == null) {
            return Result.failed(MessageFormat.format(Status.PROCESS_INSTANCE_NOT_EXIST.getMsg(),processInstanceId));
        }
        //检查用户是否有权限
        if (!checkOperationPermission(user, processInstance)) return Result.failed("操作失败，用户权限不足");

        ProcessDefinition processDefinition = processService.findProcessDefinition(processInstance.getProcessDefinitionCode(),
                processInstance.getProcessDefinitionVersion());
        Result result=null;
        if (executeType != ExecuteType.STOP && executeType != ExecuteType.PAUSE) {
             result = checkProcessDefinitionValid(processDefinition);
            if (result.getResp_code() != CodeEnum.SUCCESS.getCode()) {
                return result;
            }
        }

         result = checkExecuteType(processInstance, executeType);
        if (result.getResp_code() != CodeEnum.SUCCESS.getCode()) {
            return result;
        }

        //get the startParams user specified at the first starting while repeat running is needed
        Map<String, Object> commandMap = JSONUtils.parseObject(processInstance.getCommandParam(), new TypeReference<Map<String, Object>>() {});
        String startParams = null;
        if (MapUtils.isNotEmpty(commandMap) && executeType == ExecuteType.REPEAT_RUNNING) {
            Object startParamsJson = commandMap.get(Constants.CMD_PARAM_START_PARAMS);
            if (startParamsJson != null) {
                startParams = startParamsJson.toString();
            }
        }

        switch (executeType) {
            case REPEAT_RUNNING:
                result = insertCommand(user, processInstanceId, processDefinition.getCode(), processDefinition.getVersion(), CommandType.REPEAT_RUNNING, startParams);
                break;
            case RECOVER_SUSPENDED_PROCESS:
                result = insertCommand(user, processInstanceId, processDefinition.getCode(), processDefinition.getVersion(), CommandType.RECOVER_SUSPENDED_PROCESS, startParams);
                break;
            case START_FAILURE_TASK_PROCESS:
                result = insertCommand(user, processInstanceId, processDefinition.getCode(), processDefinition.getVersion(), CommandType.START_FAILURE_TASK_PROCESS, startParams);
                break;
            case STOP:
                if (processInstance.getState() == ExecutionStatus.READY_STOP) {
                    result=Result.failed(MessageFormat.format(Status.PROCESS_INSTANCE_ALREADY_CHANGED.getMsg(), processInstance.getName(), processInstance.getState()));
                } else {
                    result = updateProcessInstancePrepare(processInstance, CommandType.STOP, ExecutionStatus.READY_STOP);
                }
                break;
            case PAUSE:
                if (processInstance.getState() == ExecutionStatus.READY_PAUSE) {
                    result=Result.failed(MessageFormat.format(Status.PROCESS_INSTANCE_ALREADY_CHANGED.getMsg(), processInstance.getName(), processInstance.getState()));
                } else {
                    result = updateProcessInstancePrepare(processInstance, CommandType.PAUSE, ExecutionStatus.READY_PAUSE);
                }
                break;
            default:
                logger.error("unknown execute type : {}", executeType);
                result=Result.failed(MessageFormat.format(Status.REQUEST_PARAMS_NOT_VALID_ERROR.getMsg(), "unknown execute type"));
                break;
        }
        return result;
    }

    /**
     * prepare to update process instance command type and status
     *
     * @param processInstance process instance
     * @param commandType command type
     * @param executionStatus execute status
     * @return update result
     */
    private Result updateProcessInstancePrepare(ProcessInstance processInstance, CommandType commandType, ExecutionStatus executionStatus) {
        Map<String, Object> result = new HashMap<>();

        processInstance.setCommandType(commandType);
        processInstance.addHistoryCmd(commandType);
        processInstance.setState(executionStatus);
        int update = processService.updateProcessInstance(processInstance);

        // determine whether the process is normal
        if (update > 0) {
            String host = processInstance.getHost();
            String address = host.split(":")[0];
            int port = Integer.parseInt(host.split(":")[1]);
            StateEventChangeCommand stateEventChangeCommand = new StateEventChangeCommand(
                    processInstance.getId(), 0l, processInstance.getState(), processInstance.getId(), 0l
            );
            stateEventCallbackService.sendResult(address, port, stateEventChangeCommand.convert2Command());
            return Result.succeed("操作成功");
        } else {
            return Result.failed(Status.EXECUTE_PROCESS_INSTANCE_ERROR.getMsg());
        }
    }

    /**
     * insert command, used in the implementation of the page, re run, recovery (pause / failure) execution
     *
     * @param user           login user
     * @param instanceId          instance id
     * @param processDefinitionCode process definition code
     * @param commandType         command type
     * @return insert result code
     */
    private Result insertCommand(SysUser user, Long instanceId, Long processDefinitionCode, Integer processVersion, CommandType commandType, String startParams) {
        Result result = null;

        //To add startParams only when repeat running is needed
        Map<String, Object> cmdParam = new HashMap<>();
        cmdParam.put(CMD_PARAM_RECOVER_PROCESS_ID_STRING, instanceId);
        if (!StringUtils.isEmpty(startParams)) {
            cmdParam.put(CMD_PARAM_START_PARAMS, startParams);
        }

        Command command = new Command();
        command.setCommandType(commandType);
        command.setProcessDefinitionCode(processDefinitionCode);
        command.setCommandParam(JSONUtils.toJsonString(cmdParam));
        command.setUserId(user.getId());
        command.setUsername(user.getUsername());
        command.setProId(user.getCurrentProjectId());
        command.setProName(user.getProName());
        command.setProcessDefinitionVersion(processVersion);
        command.setProcessInstanceId(instanceId);

        if (!processService.verifyIsNeedCreateCommand(command)) {
            result=Result.failed(MessageFormat.format(Status.PROCESS_INSTANCE_EXECUTING_COMMAND.getMsg(), processDefinitionCode));
            return result;
        }

        int create = processService.createCommand(command);

        if (create > 0) {
            result=Result.succeed("操作成功");
        } else {
            result=Result.failed( Status.EXECUTE_PROCESS_INSTANCE_ERROR.getMsg());
        }

        return result;
    }

    /**
     * Check the state of process instance and the type of operation match
     *
     * @param processInstance process instance
     * @param executeType execute type
     * @return check result code
     */
    private Result checkExecuteType(ProcessInstance processInstance, ExecuteType executeType) {
        ExecutionStatus executionStatus = processInstance.getState();
        boolean checkResult = false;
        switch (executeType) {
            case PAUSE:
            case STOP:
                if (executionStatus.typeIsRunning()) {
                    checkResult = true;
                }
                break;
            case REPEAT_RUNNING:
                if (executionStatus.typeIsFinished()) {
                    checkResult = true;
                }
                break;
            case START_FAILURE_TASK_PROCESS:
                if (executionStatus.typeIsFailure()) {
                    checkResult = true;
                }
                break;
            case RECOVER_SUSPENDED_PROCESS:
                if (executionStatus.typeIsPause() || executionStatus.typeIsCancel()) {
                    checkResult = true;
                }
                break;
            default:
                break;
        }
        if (!checkResult) {
            return Result.failed(MessageFormat.format(Status.PROCESS_INSTANCE_STATE_OPERATION_ERROR.getMsg(), processInstance.getName(), executionStatus.toString(), executeType.toString()));
        } else {
            return Result.succeed("操作成功");
        }
    }


    /**
     * create command
     *
     * @param user user
     * @param commandType commandType
     * @param processDefinitionCode processDeCode
     * @param nodeDep nodeDep
     * @param failureStrategy failureStrategy
     * @param startNodeList startNodeList
     * @param schedule schedule
     * @param warningType warningType
     * @param warningGroupId warningGroupId
     * @param runMode runMode
     * @param processInstancePriority processInstancePriority
     * @return command id
     */
    private int createCommand(SysUser user,CommandType commandType, Long processDefinitionCode,
                              TaskDependType nodeDep, FailureStrategy failureStrategy,
                              String startNodeList, String schedule, WarningType warningType,
                              Long warningGroupId, RunMode runMode, Priority processInstancePriority,
                              Map<String, String> startParams, Integer expectedParallelismNumber, Integer dryRun) {

        /**
         * instantiate command schedule instance
         */
        Command command = new Command();

        Map<String, String> cmdParam = new HashMap<>();
        if (commandType == null) {
            command.setCommandType(CommandType.START_PROCESS);
        } else {
            command.setCommandType(commandType);
        }
        command.setProcessDefinitionCode(processDefinitionCode);
        if (nodeDep != null) {
            command.setTaskDependType(nodeDep);
        }
        if (failureStrategy != null) {
            command.setFailureStrategy(failureStrategy);
        }

        if (!StringUtils.isEmpty(startNodeList)) {
            cmdParam.put(CMD_PARAM_START_NODE_NAMES, startNodeList);
        }
        if (warningType != null) {
            command.setWarningType(warningType);
        }
        if (startParams != null && startParams.size() > 0) {
            cmdParam.put(CMD_PARAM_START_PARAMS, JSONUtils.toJsonString(startParams));
        }
        command.setCommandParam(JSONUtils.toJsonString(cmdParam));
        command.setUserId(user.getId());
        command.setUsername(user.getUsername());
        command.setProId(user.getCurrentProjectId());
        command.setProName(user.getProName());
        command.setWarningGroupId(warningGroupId);
        command.setProcessInstancePriority(processInstancePriority);
        command.setDryRun(dryRun);
        ProcessDefinition processDefinition = processDefinitionMapper.queryByCode(processDefinitionCode,user.getCurrentProjectId());
        if (processDefinition != null) {
            command.setProcessDefinitionVersion(processDefinition.getVersion());
        }

        Date start = null;
        Date end = null;
        if (!StringUtils.isEmpty(schedule)) {
            String[] interval = schedule.split(",");
            if (interval.length == 2) {
                start = DateUtils.getScheduleDate(interval[0]);
                end = DateUtils.getScheduleDate(interval[1]);
                if (start.after(end)) {
                    logger.info("complement data error, wrong date start:{} and end date:{} ",
                            start, end
                    );
                    return 0;
                }
            }
        }
        // determine whether to complement
        if (commandType == CommandType.COMPLEMENT_DATA) {
            if (start == null || end == null) {
                return 0;
            }
            return createComplementCommandList(start, end, runMode, command, expectedParallelismNumber);
        } else {
            command.setCommandParam(JSONUtils.toJsonString(cmdParam));
            return processService.createCommand(command);
        }
    }

    /**
     * create complement command
     * close left open right
     *
     * @param start
     * @param end
     * @param runMode
     * @return
     */
    private int createComplementCommandList(Date start, Date end, RunMode runMode, Command command, Integer expectedParallelismNumber) {
        int createCount = 0;
        runMode = (runMode == null) ? RunMode.RUN_MODE_SERIAL : runMode;
        Map<String, String> cmdParam = JSONUtils.toMap(command.getCommandParam());
        switch (runMode) {
            case RUN_MODE_SERIAL: {
                cmdParam.put(CMDPARAM_COMPLEMENT_DATA_START_DATE, DateUtils.dateToString(start));
                cmdParam.put(CMDPARAM_COMPLEMENT_DATA_END_DATE, DateUtils.dateToString(end));
                command.setCommandParam(JSONUtils.toJsonString(cmdParam));
                createCount = processService.createCommand(command);
                break;
            }
            case RUN_MODE_PARALLEL: {
                LinkedList<Date> listDate = new LinkedList<>();
                List<Schedule> schedules = processService.queryReleaseSchedulerListByProcessDefinitionCode(command.getProcessDefinitionCode());
                listDate.addAll(CronUtils.getSelfFireDateList(start, end, schedules));
                createCount = listDate.size();
                if (!CollectionUtils.isEmpty(listDate)) {
                    if (expectedParallelismNumber != null && expectedParallelismNumber != 0) {
                        createCount = Math.min(listDate.size(), expectedParallelismNumber);
                    }
                    logger.info("In parallel mode, current expectedParallelismNumber:{}", createCount);

                    listDate.addLast(end);
                    int chunkSize = listDate.size() / createCount;

                    for (int i = 0; i < createCount; i++) {
                        int rangeStart = i == 0 ? i : (i * chunkSize);
                        int rangeEnd = i == createCount - 1 ? listDate.size() - 1
                                : rangeStart + chunkSize;

                        cmdParam.put(CMDPARAM_COMPLEMENT_DATA_START_DATE, DateUtils.dateToString(listDate.get(rangeStart)));
                        cmdParam.put(CMDPARAM_COMPLEMENT_DATA_END_DATE, DateUtils.dateToString(listDate.get(rangeEnd)));
                        command.setCommandParam(JSONUtils.toJsonString(cmdParam));
                        processService.createCommand(command);
                    }
                }
                break;
            }
            default:
                break;
        }
        logger.info("create complement command count: {}", createCount);
        return createCount;
    }

    /**
     * check whether master exists
     *
     * @return master exists return true , otherwise return false
     */
    private boolean checkMasterExists() {
        // check master server exists
        List<Server> masterServers = monitorService.getServerListFromRegistry(true);

        // no master
        if (masterServers.isEmpty()) {
            return false;
        }
        return true;
    }

    /**
     * 只有项目管理员和作业流执行人拥有修改权限
     *
     * @param user
     * @param processInstance
     * @return
     */
    private boolean checkOperationPermission(SysUser user, ProcessInstance processInstance) {
        if (user.getCurrentProjectId().equals(processInstance.getProId()) && UserRoleUtil.checkProAdmin(user.getProRoles()))
            return true;
        if (processInstance.getUsername().equals(user.getUsername())) return true;
        return false;
    }

}
